package drds.binlog.client.impl;

import drds.binlog.client.Connector;
import drds.binlog.client.NodeAccessStrategy;
import drds.binlog.protocol.Message;
import drds.binlog.protocol.exception.ClientException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.SocketAddress;
import java.util.concurrent.TimeUnit;

/**
 * 集群版本connector实现，自带了failover功能<br/>
 */
public class ClusterConnector implements Connector
{

    private final Logger logger = LoggerFactory.getLogger(this.getClass());
    private String username;
    private String password;
    private int soTimeout = 60000;
    private int idleTimeout = 60 * 60 * 1000;
    private int retryTimes = 3;                                       // 设置-1时可以subscribe阻塞等待时优雅停机
    private int retryInterval = 5000;                                    // 重试的时间间隔，默认5秒
    private NodeAccessStrategy nodeAccessStrategy;
    private SimpleConnector simpleConnector;
    private String destination;
    private String filter;                                                  // 记录上一次的filter提交值,便于自动重试时提交

    public ClusterConnector(String username, String password, String destination,
                            NodeAccessStrategy nodeAccessStrategy)
    {
        this.username = username;
        this.password = password;
        this.destination = destination;
        this.nodeAccessStrategy = nodeAccessStrategy;
    }

    public void connect() throws ClientException
    {
        while (simpleConnector == null)
        {
            int times = 0;
            while (true)
            {
                try
                {
                    simpleConnector = new SimpleConnector(null, username, password, destination)
                    {

                        @Override
                        public SocketAddress getNextAddress()
                        {
                            return nodeAccessStrategy.nextNode();
                        }

                    };
                    simpleConnector.setSoTimeout(soTimeout);
                    simpleConnector.setIdleTimeout(idleTimeout);
                    if (filter != null)
                    {
                        simpleConnector.setFilter(filter);
                    }
                    if (nodeAccessStrategy instanceof ClusterNodeAccessStrategy)
                    {
                        simpleConnector.setZooKeeperClient(((ClusterNodeAccessStrategy) nodeAccessStrategy).getZooKeeperClient());
                    }

                    simpleConnector.connect();
                    break;
                } catch (Exception e)
                {
                    logger.warn("failed to connect to:{} after retry {} times", nodeAccessStrategy.currentNode(), times);
                    simpleConnector.disconnect();
                    simpleConnector = null;
                    // retry for #retryTimes for each node when trying to
                    // connect to it.
                    times = times + 1;
                    if (times >= retryTimes)
                    {
                        throw new ClientException(e);
                    } else
                    {
                        // fixed issue #55，增加sleep控制，避免重试connect时cpu使用过高
                        try
                        {
                            Thread.sleep(retryInterval);
                        } catch (InterruptedException e1)
                        {
                            throw new ClientException(e1);
                        }
                    }
                }
            }
        }
    }

    public boolean checkValid()
    {
        return simpleConnector != null && simpleConnector.checkValid();
    }

    public void disconnect() throws ClientException
    {
        if (simpleConnector != null)
        {
            simpleConnector.disconnect();
            simpleConnector = null;
        }
    }

    public void subscribe() throws ClientException
    {
        subscribe(""); // 传递空字符即可
    }

    public void subscribe(String filter) throws ClientException
    {
        int times = 0;
        while (times < retryTimes)
        {
            try
            {
                simpleConnector.subscribe(filter);
                this.filter = filter;
                return;
            } catch (Throwable t)
            {
                if (retryTimes == -1 && t.getCause() instanceof InterruptedException)
                {
                    logger.info("block waiting interrupted by other thread.");
                    return;
                } else
                {
                    logger.warn(String.format("something goes wrong when subscribing from server: %s",
                            simpleConnector != null ? simpleConnector.getSocketAddress() : "null"), t);
                    times++;
                    restart();
                    logger.info("restart the connector for next round retry.");
                }

            }
        }

        throw new ClientException("failed to subscribe after " + times + " times retry.");
    }

    public void unsubscribe() throws ClientException
    {
        int times = 0;
        while (times < retryTimes)
        {
            try
            {
                simpleConnector.unsubscribe();
                return;
            } catch (Throwable t)
            {
                logger.warn(String.format("something goes wrong when unsubscribing from server:%s",
                        simpleConnector != null ? simpleConnector.getSocketAddress() : "null"), t);
                times++;
                restart();
                logger.info("restart the connector for next round retry.");
            }
        }
        throw new ClientException("failed to unsubscribe after " + times + " times retry.");
    }

    public Message get(int batchSize) throws ClientException
    {
        int times = 0;
        while (times < retryTimes)
        {
            try
            {
                Message message = simpleConnector.get(batchSize);
                return message;
            } catch (Throwable t)
            {
                logger.warn(String.format("something goes wrong when getting data from server:%s",
                        simpleConnector != null ? simpleConnector.getSocketAddress() : "null"), t);
                times++;
                restart();
                logger.info("restart the connector for next round retry.");
            }
        }
        throw new ClientException("failed to fetch the data after " + times + " times retry");
    }

    public Message get(int batchSize, Long timeout, TimeUnit unit) throws ClientException
    {
        int times = 0;
        while (times < retryTimes)
        {
            try
            {
                Message msg = simpleConnector.get(batchSize, timeout, unit);
                return msg;
            } catch (Throwable t)
            {
                logger.warn(String.format("something goes wrong when getting data from server:%s",
                        simpleConnector != null ? simpleConnector.getSocketAddress() : "null"), t);
                times++;
                restart();
                logger.info("restart the connector for next round retry.");
            }
        }
        throw new ClientException("failed to fetch the data after " + times + " times retry");
    }

    public Message getWithoutAck(int batchSize) throws ClientException
    {
        int times = 0;
        while (times < retryTimes)
        {
            try
            {
                Message msg = simpleConnector.getWithoutAck(batchSize);
                return msg;
            } catch (Throwable t)
            {
                logger.warn(String.format("something goes wrong when getWithoutAck data from server:%s",
                        simpleConnector != null ? simpleConnector.getSocketAddress() : "null"), t);
                times++;
                restart();
                logger.info("restart the connector for next round retry.");
            }
        }
        throw new ClientException("failed to fetch the data after " + times + " times retry");
    }

    public Message getWithoutAck(int batchSize, Long timeout, TimeUnit unit) throws ClientException
    {
        int times = 0;
        while (times < retryTimes)
        {
            try
            {
                Message message = simpleConnector.getWithoutAck(batchSize, timeout, unit);
                return message;
            } catch (Throwable t)
            {
                logger.warn(String.format("something goes wrong when getWithoutAck data from server:%s",
                        simpleConnector != null ? simpleConnector.getSocketAddress() : "null"), t);
                times++;
                restart();
                logger.info("restart the connector for next round retry.");
            }
        }
        throw new ClientException("failed to fetch the data after " + times + " times retry");
    }

    public void rollback(long batchId) throws ClientException
    {
        int times = 0;
        while (times < retryTimes)
        {
            try
            {
                simpleConnector.rollback(batchId);
                return;
            } catch (Throwable t)
            {
                logger.warn(String.format("something goes wrong when rollbacking data from server:%s",
                        simpleConnector != null ? simpleConnector.getSocketAddress() : "null"),
                        t);
                times++;
                restart();
                logger.info("restart the connector for next round retry.");
            }
        }
        throw new ClientException("failed to rollback after " + times + " times retry");
    }

    public void rollback() throws ClientException
    {
        int times = 0;
        while (times < retryTimes)
        {
            try
            {
                simpleConnector.rollback();
                return;
            } catch (Throwable t)
            {
                logger.warn(String.format("something goes wrong when rollbacking data from server:%s",
                        simpleConnector != null ? simpleConnector.getSocketAddress() : "null"),
                        t);
                times++;
                restart();
                logger.info("restart the connector for next round retry.");
            }
        }

        throw new ClientException("failed to rollback after " + times + " times retry");
    }

    public void ack(long batchId) throws ClientException
    {
        int times = 0;
        while (times < retryTimes)
        {
            try
            {
                simpleConnector.ack(batchId);
                return;
            } catch (Throwable t)
            {
                logger.warn(String.format("something goes wrong when acking data from server:%s",
                        simpleConnector != null ? simpleConnector.getSocketAddress() : "null"),
                        t);
                times++;
                restart();
                logger.info("restart the connector for next round retry.");
            }
        }

        throw new ClientException("failed to ack after " + times + " times retry");
    }

    private void restart() throws ClientException
    {
        disconnect();
        try
        {
            Thread.sleep(retryInterval);
        } catch (InterruptedException e)
        {
            throw new ClientException(e);
        }
        connect();
    }

    // ============================= setter / getter
    // ============================

    public String getUsername()
    {
        return username;
    }

    public void setUsername(String username)
    {
        this.username = username;
    }

    public String getPassword()
    {
        return password;
    }

    public void setPassword(String password)
    {
        this.password = password;
    }

    public int getSoTimeout()
    {
        return soTimeout;
    }

    public void setSoTimeout(int soTimeout)
    {
        this.soTimeout = soTimeout;
    }

    public int getIdleTimeout()
    {
        return idleTimeout;
    }

    public void setIdleTimeout(int idleTimeout)
    {
        this.idleTimeout = idleTimeout;
    }

    public int getRetryTimes()
    {
        return retryTimes;
    }

    public void setRetryTimes(int retryTimes)
    {
        this.retryTimes = retryTimes;
    }

    public int getRetryInterval()
    {
        return retryInterval;
    }

    public void setRetryInterval(int retryInterval)
    {
        this.retryInterval = retryInterval;
    }

    public NodeAccessStrategy getNodeAccessStrategy()
    {
        return nodeAccessStrategy;
    }

    public void setNodeAccessStrategy(NodeAccessStrategy nodeAccessStrategy)
    {
        this.nodeAccessStrategy = nodeAccessStrategy;
    }

    public SimpleConnector getSimpleConnector()
    {
        return simpleConnector;
    }

    public void stopRunning()
    {
        simpleConnector.stopRunning();
    }

}
